package com.imooc.imoocbilibili.socket;

import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.imooc.imoocbilibili.constants.MQConstant;
import com.imooc.imoocbilibili.controller.WebsocketTestController;
import com.imooc.imoocbilibili.entity.Danmu;
import com.imooc.imoocbilibili.interceptor.ThreadUser;
import com.imooc.imoocbilibili.service.DanmuService;
import com.imooc.imoocbilibili.utils.RocketMQUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author 江黎
 * @since 2022-05-15
 */
@Component
@ServerEndpoint("/webSocket/{key}")
@Slf4j
public class WebSocket {

    private Session session;
    public static final ConcurrentHashMap<String, WebSocket> webSocketMap = new ConcurrentHashMap<>();

    @OnOpen
    public void onOpen(Session session, @PathParam(value = "key") String key) {
        // 测试:WebSocket是多例的,不能通过@Autowired注入bean
//        WebsocketTestController websocketTestController = SpringUtil.getBean("websocketTestController");
//        System.out.println(websocketTestController);
        this.session = session;
        webSocketMap.put(key, this);
        log.info("【websocket消息】有新的连接, 总数:{}", webSocketMap.size());
    }

    @OnClose
    public void onClose(@PathParam(value = "key") String key) {
        webSocketMap.remove(key);
        log.info("【websocket消息】连接断开, 总数:{}", webSocketMap.size());
    }

    @OnMessage
    public void onMessage(String message, @PathParam(value = "key") String key) {
        log.info("【websocket消息】收到客户端发来的消息:{}", message);
        // 收到消息后,服务器端可以做其他操作
        this.sendMessage(message);
    }

    /**
     * 指定用户发送
     */
    public void sendMessage(String message, String key) {
        log.info("【websocket消息】发送指定消息, message={}, key={}", message, key);
        try {
            webSocketMap.get(key).session.getBasicRemote().sendText(message);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 群发
     */
    public void sendMessage(String message) {
        log.info("【websocket消息】广播消息, message={}", message);
        for (String key : webSocketMap.keySet()) {
            try {
                // 发送弹幕
//                webSocketMap.get(key).session.getBasicRemote().sendText(message);

                // 发送弹幕改造: 用MQ异步
                DefaultMQProducer danmuProducer = SpringUtil.getBean("danmuProducer");
                JSONObject jsonObject = new JSONObject();
                jsonObject.set("message", message);
                jsonObject.set("key", key);
                Message msg = new Message(MQConstant.DANMU_TOPIC, JSONUtil.toJsonStr(jsonObject).getBytes(StandardCharsets.UTF_8));
                RocketMQUtil.asyncSendMsg(danmuProducer, msg);

                // 保存弹幕
                if (ThreadUser.getUser() != null) {
                    Danmu danmu = JSONUtil.toBean(message, Danmu.class);
                    danmu.setUserid(ThreadUser.getUser().getId());
                    danmu.setCreatetime(new Date());
                    DanmuService danmuService = SpringUtil.getBean("danmuService");
                    // 可以用Async异步提高性能
                    danmuService.save(danmu);
                    // 保存到redis
                    danmuService.saveDanmuToRedis(danmu);
                }
            } catch (Exception e) {
                log.error("弹幕发送失败");
                e.printStackTrace();
            }
        }
    }

    public Session getSession() {
        return this.session;
    }

    // 每隔5秒向前端推送在线人数
    @Scheduled(fixedRate = 5000)
    private void noticeOnlineCount() throws IOException {
        for (Map.Entry<String, WebSocket> entry : webSocketMap.entrySet()) {
            WebSocket webSocket = entry.getValue();
            if (webSocket.session.isOpen()) {
                JSONObject jsonObject = new JSONObject();
                jsonObject.set("onlineCount", webSocketMap.size());
                webSocket.session.getBasicRemote().sendText(JSONUtil.toJsonStr(jsonObject));
            }
        }
    }
}
